package com.navinfo.opentsp.platform.computing.serializer;

import com.alibaba.fastjson.JSON;
import com.google.common.base.Charsets;
import com.google.common.collect.Lists;
import com.navinfo.opentsp.platform.computing.dto.*;
import com.navinfo.opentsp.platform.computing.util.Base64;
import com.navinfo.opentsp.platform.computing.util.DateUtils;
import com.navinfo.opentsp.platform.location.protocol.common.LCLocationData;
import com.navinfo.opentsp.platform.location.protocol.common.LCVehicleBreakdown;
import com.navinfo.opentsp.platform.location.protocol.common.LCVehicleStatusData;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.serialization.EventSerializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.io.OutputStream;
import java.util.List;

/**
* locationData信息
* HDFS_Sink的locationData序列化器，拼接成字段间通过\001分割的字符串
*/
public class HDFSLocationDataSerializer implements EventSerializer {
    private static final Logger logger = LoggerFactory.getLogger(HDFSLocationDataSerializer.class);

    private final String APPEND_NEWLINE = "appendNewline";
    private final boolean appendNewline;
    private final String splitStr = "\001";
    private final OutputStream out;

    private HDFSLocationDataSerializer(OutputStream out, Context ctx) {
        this.appendNewline = ctx.getBoolean(APPEND_NEWLINE, true);
        this.out = out;
    }

    @Override
    public boolean supportsReopen() {
        return false;
    }
    @Override
    public void afterCreate() {}
    @Override
    public void afterReopen() {}
    @Override
    public void beforeClose() {}
    @Override
    public void flush() throws IOException {
        this.out.flush();
    }
    @Override
    public void write(Event event) throws IOException {
        try {
            //拼写落盘到HDFS的字符串。字段间分隔符是\001，null通过\N表示
            LCLocationData.LocationData locationData = LCLocationData.LocationData.parseFrom(event.getBody());

            StringBuilder locationDataStr = new StringBuilder(5000);
                    locationDataStr.append(locationData.getAlarm()).append(splitStr);
                    locationDataStr.append(locationData.getStatus()).append(splitStr);
                    locationDataStr.append(locationData.getLongitude()).append(splitStr);
                    locationDataStr.append(locationData.getLatitude()).append(splitStr);
                    locationDataStr.append(locationData.getOriginalLng()).append(splitStr);
                    locationDataStr.append(locationData.getOriginalLat()).append(splitStr);
                    locationDataStr.append(locationData.getHeight()).append(splitStr);
                    locationDataStr.append(locationData.getSpeed()).append(splitStr);
                    locationDataStr.append(locationData.getDirection()).append(splitStr);
                    locationDataStr.append(locationData.getGpsDate()).append(splitStr);
                    locationDataStr.append(locationData.getMileage()).append(splitStr);
                    locationDataStr.append(locationData.getReceiveDate()).append(splitStr);
                    locationDataStr.append(locationData.getIsPatch()).append(splitStr);
                    locationDataStr.append(locationData.getOil()).append(splitStr);
                    locationDataStr.append(locationData.getRecorderSpeed()).append(splitStr);
                    locationDataStr.append(JSON.toJSON(getSpeedAdditions(locationData.getSpeedAdditionList()))).append(splitStr);
                    locationDataStr.append(JSON.toJSON(getAreaAdditions(locationData.getAreaAdditionList()))).append(splitStr);
                    locationDataStr.append(JSON.toJSON(getRouteAdditions(locationData.getRouteAdditionList()))).append(splitStr);
                    locationDataStr.append(JSON.toJSON(getTemAlarmDtos(locationData.getTemAlarmList()))).append(splitStr);
                    locationDataStr.append(locationData.getStarStatus()).append(splitStr);
                    locationDataStr.append(locationData.getStarNumber()).append(splitStr);
                    locationDataStr.append(JSON.toJSON(locationData.getAlarmIdentifyList())).append(splitStr);
                    locationDataStr.append(locationData.getSignalStatus()).append(splitStr);
                    locationDataStr.append(locationData.getIoStatus()).append(splitStr);
                    locationDataStr.append(locationData.getAnalogAD0()).append(splitStr);
                    locationDataStr.append(locationData.getAnalogAD1()).append(splitStr);
                    locationDataStr.append(locationData.getSignalStrength()).append(splitStr);
                    locationDataStr.append(locationData.getSatelliteNumber()).append(splitStr);
                    locationDataStr.append(locationData.getIsValid()).append(splitStr);
                    locationDataStr.append(JSON.toJSON(getKeyPointFences(locationData.getDefenceAdditionList()))).append(splitStr);
                    locationDataStr.append(JSON.toJSON(getParkingAdditions(locationData.getParkingAdditionList()))).append(splitStr);
                    locationDataStr.append(JSON.toJSON(getVehicleStatusDataString(locationData.getStatusAddition().getStatusList()))).append(splitStr);
                    locationDataStr.append(JSON.toJSON(getVehicleBreakdowns(locationData.getCancleBreakdownAddition().getBreakdownList()))).append(splitStr);
                    locationDataStr.append(JSON.toJSON(locationData.getAdditionAlarm().toByteArray())).append(splitStr);

                    locationDataStr.append(locationData.getAnalysisData().getTurningAngle()).append(splitStr)
                            .append(locationData.getAnalysisData().getLowOilDrivingValue()).append(splitStr)
                            .append(locationData.getAnalysisData().getVehicleSpeedFromEcu()).append(splitStr)
                            .append(locationData.getAnalysisData().getRpmWhenAlarming()).append(splitStr)
                            .append(locationData.getAnalysisData().getVelocityChangeValue()).append(splitStr)
                            .append(locationData.getAnalysisData().getCurrentGearshift()).append(splitStr)
                            .append(locationData.getAnalysisData().getCurrentRPM()).append(splitStr)
                            .append(locationData.getAnalysisData().getBrakeTimes()).append(splitStr)
                            .append(locationData.getAnalysisData().getClutchTimes()).append(splitStr)
                            .append(locationData.getAnalysisData().getRetarderTimes()).append(splitStr)
                            .append(locationData.getAnalysisData().getAbsTimes()).append(splitStr)
                            .append(locationData.getAnalysisData().getReverseTimes()).append(splitStr)
                            .append(Base64.encode(locationData.getAnalysisData().getDeviceIdentity().getBytes("GBK"))).append(splitStr);

                    locationDataStr.append(locationData.getBatteryPower()).append(splitStr);
                    locationDataStr.append(JSON.toJSON(getModuleVoltagess(locationData.getModuleVoltagesList()))).append(splitStr);
                    locationDataStr.append(locationData.getElectricVehicle()).append(splitStr);

                    locationDataStr.append(locationData.getBatteryInfo().getMotorTemperature()).append(splitStr)
                            .append(locationData.getBatteryInfo().getMotorControlTemp()).append(splitStr)
                            .append(locationData.getBatteryInfo().getMcuFaultCode()).append(splitStr)
                            .append(locationData.getBatteryInfo().getBmsStatus()).append(splitStr)
                            .append(locationData.getBatteryInfo().getBatteryAveTem()).append(splitStr)
                            .append(locationData.getBatteryInfo().getBatteryCurrent()).append(splitStr)
                            .append(locationData.getBatteryInfo().getTotalBatteryV()).append(splitStr)
                            .append(locationData.getBatteryInfo().getBatteryFaultCode()).append(splitStr)
                            .append(locationData.getBatteryInfo().getBatteryMaxTem()).append(splitStr)
                            .append(locationData.getBatteryInfo().getBatteryMinTem()).append(splitStr)
                            .append(locationData.getBatteryInfo().getBatteryMinSoc()).append(splitStr)
                            .append(locationData.getBatteryInfo().getMaxChargingV()).append(splitStr)
                            .append(locationData.getBatteryInfo().getMaxChargingE()).append(splitStr)
                            .append(locationData.getBatteryInfo().getTotalStatus()).append(splitStr)
                            .append(locationData.getBatteryInfo().getVehicleSpeed()).append(splitStr)
                            .append(locationData.getBatteryInfo().getShiftStatus()).append(splitStr)
                            .append(locationData.getBatteryInfo().getVcuFaultCode()).append(splitStr)
                            .append(locationData.getBatteryInfo().getEnduranceMileage()).append(splitStr)
                            .append(locationData.getBatteryInfo().getAcceleratorPedal()).append(splitStr)
                            .append(locationData.getBatteryInfo().getBrakePedal()).append(splitStr)
                            .append(locationData.getBatteryInfo().getModelInfo()).append(splitStr)
                            .append(locationData.getBatteryInfo().getSwitchInfo()).append(splitStr)
                            .append(locationData.getBatteryInfo().getElectricAttachment()).append(splitStr)
                            .append(locationData.getBatteryInfo().getMotorRpm()).append(splitStr);

                    locationDataStr.append(locationData.getAlarmFilter()).append(splitStr);
                    locationDataStr.append(locationData.getStandardMileage()).append(splitStr);
                    locationDataStr.append(locationData.getStandardFuelCon()).append(splitStr);
                    locationDataStr.append(JSON.toJSON(getStaytimeParkingDtos(locationData.getStaytimeParkingAdditionList()))).append(splitStr);
                    locationDataStr.append(locationData.getEcuDate()).append(splitStr);

                    locationDataStr.append(locationData.getTemperatureAddition().getVehType()).append(splitStr)
                            .append(locationData.getTemperatureAddition().getTyrePosition()).append(splitStr)
                            .append(locationData.getTemperatureAddition().getTyreCondition()).append(splitStr)
                            .append(locationData.getTemperatureAddition().getTyrePressure()).append(splitStr)
                            .append(locationData.getTemperatureAddition().getTyreTemperature()).append(splitStr)
                            .append(locationData.getTemperatureAddition().getTemAlarmThreshold()).append(splitStr)
                            .append(locationData.getTemperatureAddition().getTyreHalarmThreshold()).append(splitStr)
                            .append(locationData.getTemperatureAddition().getTyreLalarmThreshold()).append(splitStr)
                            .append(locationData.getTemperatureAddition().getTyreNomimalValue()).append(splitStr);

                    locationDataStr.append(JSON.toJSON(getVehicleBreakdowns(locationData.getCancleBreakdownAddition().getBreakdownList()))).append(splitStr);
                    locationDataStr.append(locationData.getTerminalId()).append(splitStr);
                    locationDataStr.append(locationData.getStayTime()).append(splitStr);
                    locationDataStr.append(locationData.getSerialNumber()).append(splitStr);

                    locationDataStr.append(locationData.getCarLockEffect().getEngineSpeed()).append(splitStr)
                            .append(locationData.getCarLockEffect().getActualOutputTorque()).append(splitStr)
                            .append(locationData.getCarLockEffect().getThrottleOpening()).append(splitStr);

                    locationDataStr.append(locationData.getGearStatus()).append(splitStr);
                    locationDataStr.append(locationData.getTcuLoad()).append(splitStr);
                    locationDataStr.append(locationData.getVecuLoad()).append(splitStr);
                    locationDataStr.append(locationData.getUreaInjectionAvg()).append(splitStr);
                    locationDataStr.append(locationData.getUreaPressureAvg()).append(splitStr);
                    locationDataStr.append(locationData.getTransmissionOilTemperature()).append(splitStr);
                    locationDataStr.append(locationData.getTcuRamp()).append(splitStr);
                    locationDataStr.append(locationData.getEngineRunningTime()).append(splitStr);
                    locationDataStr.append(locationData.getDpfAshLoading()).append(splitStr);
                    locationDataStr.append(locationData.getCarbonLoad()).append(splitStr);
                    locationDataStr.append(locationData.getTotalureaInjection()).append(splitStr);
                    locationDataStr.append(locationData.getUreaLevel()).append(splitStr);
                    locationDataStr.append(locationData.getNewMeterMileage()).append(splitStr);
                    locationDataStr.append(locationData.getCargoBox().getCargoBoxTemperature1()).append(splitStr);
                    locationDataStr.append(locationData.getCargoBox().getCargoBoxTemperature2()).append(splitStr);
                    locationDataStr.append(locationData.getCargoBox().getCargoBoxTemperature3()).append(splitStr);
                    locationDataStr.append(locationData.getCargoBox().getCargoBoxTemperature4()).append(splitStr);
                    locationDataStr.append(locationData.getCargoBox().getContainerHumidity1()).append(splitStr);
                    locationDataStr.append(locationData.getCargoBox().getContainerHumidity2()).append(splitStr);
                    locationDataStr.append(locationData.getCargoBox().getContainerHumidity3()).append(splitStr);
                    locationDataStr.append(locationData.getCargoBox().getContainerHumidity4()).append(splitStr);
                    locationDataStr.append(locationData.getCargoBox().getCargoBoxDoorStatus1()).append(splitStr);
                    locationDataStr.append(locationData.getCargoBox().getCargoBoxDoorStatus2()).append(splitStr);
                    locationDataStr.append(locationData.getCargoBox().getCargoBoxDoorStatus3()).append(splitStr);
                    locationDataStr.append(locationData.getCargoBox().getSwitchStatus()).append(splitStr);
                    locationDataStr.append(locationData.getUreaTemperature()).append(splitStr);
                    locationDataStr.append(GasBoxDto.toString(locationData.getGasBox())).append(splitStr);

//                    locationDataStr.append(locationData.getCarbonLoad()).append(splitStr);

            //添加字段只能在分割线之上添加,注意时间分区这块，终端传的时候是秒还是毫秒
            //------------------------------------------分割线----------------------------------------
            locationDataStr.append(DateUtils.format(locationData.getGpsDate() * 1000, "yyyyMMdd")).append(splitStr)
                    .append(locationData.getTerminalId() % 32);

            //将数据拼接后写入hdfs
            this.out.write(locationDataStr.toString().getBytes(Charsets.UTF_8));
            if (this.appendNewline) {
                this.out.write(10);
            }
        }catch (Exception e){
            logger.error("执行HDFS的HDFSLocationDataSerializer拦截器失败：" + e.getMessage(), e);
        }
    }

    public static List<String> getSpeedAdditions(List<LCLocationData.SpeedAddition> speedAdditions){
        List<String> speedAdditionDtos = Lists.newArrayList();
        if(speedAdditions == null){
            return speedAdditionDtos;
        }
        for(LCLocationData.SpeedAddition speedAddition : speedAdditions){
            speedAdditionDtos.add(SpeedAdditionDto.parseString(speedAddition));
        }
        return speedAdditionDtos;
    }

    public static List<String> getAreaAdditions(List<LCLocationData.AreaAddition> areaAdditions){
        List<String> areaAdditionDtos = Lists.newArrayList();
        if(areaAdditions == null){
            return areaAdditionDtos;
        }
        for(LCLocationData.AreaAddition areaAddition : areaAdditions){
            areaAdditionDtos.add(AreaAdditionDto.parseString(areaAddition));
        }
        return areaAdditionDtos;
    }

    public static List<String> getRouteAdditions(List<LCLocationData.RouteAddition> routeAdditions){
        List<String> routeAdditionDtos = Lists.newArrayList();
        if(routeAdditions == null){
            return routeAdditionDtos;
        }
        for(LCLocationData.RouteAddition routeAddition : routeAdditions){
            routeAdditionDtos.add(RouteAdditionDto.parseString(routeAddition));
        }
        return routeAdditionDtos;
    }

    public static List<String> getTemAlarmDtos(List<LCLocationData.TemAddition> temAlarmList){
        List<String> temAlarmDtos = Lists.newArrayList();
        if(temAlarmList == null){
            return temAlarmDtos;
        }
        for(LCLocationData.TemAddition temAddition : temAlarmList){
            temAlarmDtos.add(TemAlarmDto.parseString(temAddition));
        }
        return temAlarmDtos;
    }

    public static List<String> getKeyPointFences(List<LCLocationData.KeyPointFenceAddition> keyPointFenceAdditionList){
        List<String> defenceAdditionDtos = Lists.newArrayList();
        if(keyPointFenceAdditionList == null){
            return defenceAdditionDtos;
        }
        for(LCLocationData.KeyPointFenceAddition keyPointFenceAddition : keyPointFenceAdditionList){
            defenceAdditionDtos.add(DefenceAdditionDto.parseString(keyPointFenceAddition));
        }
        return defenceAdditionDtos;
    }

    public static List<String> getParkingAdditions(List<LCLocationData.OvertimeParkingAddition> overtimeParkingAdditions){
        List<String> parkingAdditionDtos = Lists.newArrayList();
        if(overtimeParkingAdditions == null){
            return parkingAdditionDtos;
        }
        for(LCLocationData.OvertimeParkingAddition overtimeParkingAddition : overtimeParkingAdditions){
            parkingAdditionDtos.add(ParkingAdditionDto.parseString(overtimeParkingAddition));
        }
        return parkingAdditionDtos;
    }

    public static List<String> getModuleVoltagess(List<LCLocationData.ModuleVoltage> moduleVoltages){
        List<String> moduleVoltagesDtos = Lists.newArrayList();
        if(moduleVoltages == null){
            return moduleVoltagesDtos;
        }
        for(LCLocationData.ModuleVoltage moduleVoltage : moduleVoltages){
            moduleVoltagesDtos.add(ModuleVoltagesDto.parseString(moduleVoltage));
        }
        return moduleVoltagesDtos;
    }

    public static List<String> getStaytimeParkingDtos(List<LCLocationData.StaytimeParkingAddition> staytimeParkings){
        List<String> staytimeParkingDtos = Lists.newArrayList();
        if(staytimeParkings == null){
            return staytimeParkingDtos;
        }
        for(LCLocationData.StaytimeParkingAddition staytimeParking : staytimeParkings){
            staytimeParkingDtos.add(StaytimeParkingAdditionDto.parseString(staytimeParking));
        }
        return staytimeParkingDtos;
    }


    public static List<String> getVehicleStatusDataString(List<LCVehicleStatusData.VehicleStatusData> vehicleStatusDatas){
        List<String> vehicleStatusAdditionDtos = Lists.newArrayList();
        if(vehicleStatusDatas == null){
            return vehicleStatusAdditionDtos;
        }
        for(LCVehicleStatusData.VehicleStatusData vehicleStatusData : vehicleStatusDatas){
            vehicleStatusAdditionDtos.add(VehicleStatusAdditionDto.parseString(vehicleStatusData));
        }
        return vehicleStatusAdditionDtos;
    }

    public static List<String> getVehicleBreakdowns(List<LCVehicleBreakdown.VehicleBreakdown> vehicleBreakdowns){
        List<String> breakdownAdditionDtos = Lists.newArrayList();
        if(vehicleBreakdowns == null){
            return breakdownAdditionDtos;
        }
        for(LCVehicleBreakdown.VehicleBreakdown vehicleBreakdown : vehicleBreakdowns){
            String data = BreakdownAdditionDto.parseString(vehicleBreakdown);
            if(data == null){
                continue;
            }
            breakdownAdditionDtos.add(data);
        }
        return breakdownAdditionDtos;
    }


    public static class Builder implements EventSerializer.Builder {
        public Builder() { }
        @Override
        public EventSerializer build(Context context, OutputStream out) {
            return  new HDFSLocationDataSerializer(out, context);
        }
    }
}
